package com.chucang.shucang.common.mq.config;

import com.chucang.shucang.common.mq.constant.KafkaConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;

/**
 * @author flitsneak
 * @email flitsneak@gmail.com
 * @date 2022/9/26 13:53
 * @description
 */
@Slf4j
@AutoConfiguration
public class KafkaConfig {
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return (message, exception, consumer) -> {
            log.error("kafka消费异常！！！异常原因=====》{}异常消息内容=====》{}", exception.getMessage(), message.getPayload());
            //TODO 记录日志
            return null;
        };
    }

    @Bean
    public NewTopic initialTopic() {
        return new NewTopic(KafkaConstant.SMS_CODE_TOPIC, 1, (short) 1);
    }
}
